package com.flying.learning.si.sik.library;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.MessageHandler;

@Configuration
public class ProducerChannelConfig {
    @Bean
    public DirectChannel producerChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "producerChannel")
    public MessageHandler kafkaMsgHandler(KafkaTemplate template) {
        KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template);
        handler.setMessageKeyExpression(new LiteralExpression("kafka-integration"));
        return handler;
    }
}
